package com.niit.utils

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

/*
SparkUtils :主要是负责 Spark Context SparkSession StreamingContext 进行创建 管理
    由于我们分析的业务比较多，会用spark Context SparkSession StreamingContext
    为了避免重复创建，所以将已经创建好的spark Context SparkSession StreamingContext 放到本地线程池中，
    供后续操作的读取
 */
object SparkUtil {

  //创建三个本地线程池，就是用来存储 Spark Context SparkSession StreamingContext
  private val scLocal = new ThreadLocal[SparkContext]
  private val sparkLocal = new ThreadLocal[SparkSession]
  private val sscLocal = new ThreadLocal[StreamingContext]
  //将Spark Context SparkSession StreamingContext进行声明
  private var sc:SparkContext = _
  private var spark:SparkSession = _
  private var ssc:StreamingContext = _
  //将时间进行声明
  private var seconds:Duration = _;


  //创建 Spark Context SparkSession StreamingContext
  def createSpark(sparkConf: SparkConf,seconds:Duration = Seconds(3)):SparkContext={

    if(sc == null){

       spark = SparkSession.builder().config(sparkConf).getOrCreate()
       sc = spark.sparkContext // 利用已经创建好SparkSession 去获得  Spark Context

      putSC(sc)
      putSpark(spark)

      if(this.seconds != seconds){
        sscLocal.remove()
        ssc = new StreamingContext(sc,seconds)
        this.seconds = seconds
      }
      putSSC(ssc)

    }
    sc

  }

  //修改StreamingContext 主要重新设置时间
  def getOrCreateStreamingContext(sparkContext: SparkContext,seconds:Duration):StreamingContext={
      if(this.seconds != seconds){
        sscLocal.remove()
        ssc = new StreamingContext(sc,seconds)
        this.seconds = seconds
        putSSC(ssc)
    }
    ssc
  }


  //对 Spark Context SparkSession StreamingContext 添加到 相应的线程池当中
  private def putSC(sc:SparkContext): Unit ={
    scLocal.set(sc)
  }
  private def putSpark(spark:SparkSession): Unit ={
    sparkLocal.set(spark)
  }
  private def putSSC(ssc:StreamingContext): Unit ={
    sscLocal.set(ssc)
  }

  //对 Spark Context SparkSession StreamingContext 从相应的线程池中获取
  def takeSC():SparkContext={
    scLocal.get()
  }
  def takeSpark():SparkSession={
    sparkLocal.get()
  }
  def takeSSC():StreamingContext={
    sscLocal.get()
  }

  def clear(): Unit ={
    scLocal.remove()
    sparkLocal.remove()
    sscLocal.remove()
  }

}
